package com.chief.window;

import org.apache.commons.lang3.time.DateUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

public class SlidingCountWindowTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> streamSource = env.socketTextStream("hadoop55", 7777);

        SingleOutputStreamOperator<Tuple3<String, Long, Integer>> map = streamSource.map(new MapFunction<String, Tuple3<String, Long, Integer>>() {
            @Override
            public Tuple3<String, Long, Integer> map(String value) throws Exception {
                String[] split = value.split(",");
                return new Tuple3<>(split[0], DateUtils.parseDate(split[1], "yyyy-MM-dd HH:mm:ss").getTime(), Integer.valueOf(split[2]));
            }
        });

        map.assignTimestampsAndWatermarks(
                WatermarkStrategy.<Tuple3<String, Long, Integer>>forBoundedOutOfOrderness(Duration.ofMillis(200))
                        .withTimestampAssigner((event, t) -> event.f1))
                .keyBy(value -> value.f0)
                .countWindow(5, 2)
                .sum(2).print();


        env.execute("SlidingCountWindowTest");
    }
}
